package com.ydj.framework.redis;

import io.lettuce.core.ScriptOutputType;
import io.lettuce.core.SetArgs;
import io.lettuce.core.api.async.RedisAsyncCommands;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.RedisSerializer;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisCluster;
import redis.clients.jedis.JedisCommands;

import java.util.ArrayList;
import java.util.List;

/**
 * 分布式锁工具-redis实现
 *
 * @author wangsc
 * @time 2018年11月5日下午7:34:02
 */
class RedisDistributedLockHandlerImpl implements DistributedLockHandler {

    private static final Logger logger = LoggerFactory.getLogger(RedisDistributedLockHandlerImpl.class);
    /**
     * 1s的时长,ms
     */
    private static final int ONE_SECOND = 1000;
    /**
     * 默认的获取锁操作超时时长，单位ms,默认10秒
     */
    public static final int DEFAULT_ACQUIRE_TIMEOUT_MILLIS = Integer.getInteger("com.hori.jedis.lock.acquiry.millis", 10 * ONE_SECOND);
    /**
     * 默认的while循环内部尝试获取锁的时间间隔，单位ms,默认100ms
     */
    public static final int DEFAULT_ACQUIRY_RESOLUTION_MILLIS = Integer.getInteger("com.hori.jedis.lock.acquiry.resolution.millis", 100);

    /**
     * 依赖redisTemplate
     */
    private ExtRedisTemplate redisTemplate;

    public ExtRedisTemplate getRedisTemplate() {
        return redisTemplate;
    }

    public void setRedisTemplate(ExtRedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
    }


    /**
     * 操作redis获取全局锁
     *
     * @param tryInterval 多少ms尝试一次
     * @return true 获取成功，false获取失败
     */
    @SuppressWarnings("unchecked")
    private boolean getLock(DistributedLock newLock, long tryInterval) {
        return tryLock(newLock.getLockKey(), newLock.getUUID().toString(), newLock.getExpiryTime());
    }


    /**
     * 尝试获取全局锁
     *
     * @param lock 锁的名称
     * @return true 获取成功，false获取失败
     */
    @Override
    public boolean tryLock(DistributedLock lock) {
        return getLock(lock, DEFAULT_ACQUIRY_RESOLUTION_MILLIS);
    }


    /**
     * 释放锁
     */
    @Override
    public void releaseLock(DistributedLock lock) {
        releaseLock(lock.getLockKey(), lock.getUUID().toString());
    }

    public static final String UNLOCK_LUA;

    static {
        UNLOCK_LUA = "if redis.call(\"get\",KEYS[1]) == ARGV[1] " +
                "then " +
                "    return redis.call(\"del\",KEYS[1]) " +
                "else " +
                "    return 0 " +
                "end ";
    }

    /**
     * @param key    key
     * @param value  值 解锁时候需要判断值是否一致
     * @param expire 毫秒
     * @return
     */
    @Override
    public boolean tryLock(String key, String value, long expire) {
        return tryLock(key, value, expire, 15, 200);
    }

    @Override
    public boolean tryLock(String key, String value, long expire, int retryTimes, long sleepMillis) {
        boolean result = setRedis(key, value, expire);
        // 如果获取锁失败，按照传入的重试次数进行重试
        while ((!result) && retryTimes-- > 0) {
            try {
                logger.debug("lock failed, retrying..." + retryTimes);
                Thread.sleep(sleepMillis);
            } catch (InterruptedException e) {
                return false;
            }
            result = setRedis(key, value, expire);
        }
        return result;
    }

    @SuppressWarnings("unchecked")
    private boolean setRedis(String key, String value, long expire) {
        try {
            String result = (String) redisTemplate.execute(new RedisCallback<String>() {
                @Override
                public String doInRedis(RedisConnection connection) throws DataAccessException {
                    Object commands = connection.getNativeConnection();
                    String redisResult = null;
                    if (commands instanceof JedisCommands) {
                        // springboot 1.x 使用可以使用Jedis
                        redisResult = ((JedisCommands) (commands)).set(key, value, "NX", "PX", expire);
                    } else if (commands instanceof RedisAsyncCommands) {
                        // springboot 2.0以后需要使用lettuce
                        RedisSerializer<String> stringRedisSerializer = (RedisSerializer<String>) redisTemplate.getKeySerializer();
                        byte[] keyByte = stringRedisSerializer.serialize(key);
                        byte[] valueByte = stringRedisSerializer.serialize(value);
                        redisResult = ((RedisAsyncCommands<byte[], byte[]>) commands)
                                .getStatefulConnection()
                                .sync()
                                .set(keyByte, valueByte, SetArgs.Builder.nx().px(expire));
                    }
                    return redisResult;
                }
            });
            return !StringUtils.isEmpty(result);
        } catch (Exception e) {
            e.printStackTrace();
            logger.error("set redis occured an exception", e);
        }
        return false;
    }

    /**
     * @param key   key
     * @param value 值 解锁时候需要判断值是否一致
     * @return
     */
    @Override
    public boolean releaseLock(String key, String value) {
        // 释放锁的时候，有可能因为持锁之后方法执行时间大于锁的有效期，此时有可能已经被另外一个线程持有锁，所以不能直接删除
        try {
            List<String> keys = new ArrayList<>();
            keys.add(key);
            List<String> args = new ArrayList<>();
            args.add(value);

            // 使用lua脚本删除redis中匹配value的key，可以避免由于方法执行时间过长而redis锁自动过期失效的时候误删其他线程的锁
            // spring自带的执行脚本方法中，集群模式直接抛出不支持执行脚本的异常，所以只能拿到原redis的connection来执行脚本

            Long result = (Long) redisTemplate.execute(new RedisCallback<Long>() {
                @Override
                @SuppressWarnings("unchecked")
                public Long doInRedis(RedisConnection connection) throws DataAccessException {
                    String lua = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
                    Object nativeConnection = connection.getNativeConnection();
                    // 集群模式和单机模式虽然执行脚本的方法一样，但是没有共同的接口，所以只能分开执行
                    // 集群模式. springboot 1.x 使用可以使用Jedis
                    if (nativeConnection instanceof JedisCluster) {
                        return (Long) ((JedisCluster) nativeConnection).eval(lua, keys, args);
                    }

                    // 单机模式. springboot 1.x 使用可以使用Jedis
                    else if (nativeConnection instanceof Jedis) {
                        return (Long) ((Jedis) nativeConnection).eval(lua, keys, args);
                    }
                    // springboot 2.0以后需要使用lettuce
                    else if (nativeConnection instanceof RedisAsyncCommands) {
                        RedisSerializer<String> stringRedisSerializer = (RedisSerializer<String>) redisTemplate.getKeySerializer();
                        byte[][] keyByteArray = new byte[1][];
                        keyByteArray[0] = stringRedisSerializer.serialize(key);
                        byte[][] valueByteArray = new byte[1][];
                        valueByteArray[0] = stringRedisSerializer.serialize(value);

                        return (Long) ((RedisAsyncCommands<byte[], byte[]>) nativeConnection)
                                .getStatefulConnection()
                                .sync()
                                .eval(lua, ScriptOutputType.INTEGER, keyByteArray, valueByteArray);
                    }
                    return 0L;
                }
            });

            return result != null && result > 0;
        } catch (Exception e) {
            logger.error("release lock occured an exception", e);
        }
        return false;
    }
}
